package com.bwie.realtime.jtp.dwd.log.job;

import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class sss {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "topic-log");

        kafkaDataStream.print("kafka");

        DataStream<String> pageStream = JtpAppLogEtlJob.processLog(kafkaDataStream);

        KafkaUtil.producerKafka(pageStream, "dwd-traffic-page-log");

        env.execute("JtpAppLogEtlJob");
    }
    public static DataStream<String> processLog(DataStream<String> stream) {

        DataStream<String> jsonStream = JtpAppLogEtlJob.splitStream(stream);

        DataStream<String> isNewStream = JtpAppLogEtlJob.processIsNew(jsonStream);

        return isNewStream;
    }




























}
